/*
 *	workmanager.c
 *	© 2009 Tom Booth
 *
 *	A work manager for a set of worker threads
 *
 */

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

#include "workmanager.h"

struct wm_queue_item {
	void *payload;
	struct wm_queue_item* next;
	unsigned int is_special:1;
};

struct wm_special_thread {
	pthread_t thread;		
	struct wm_special_thread *next;
};


static void wm_free_queueitem(struct wm_queue_item *item);
static void wm_free_specialitem(struct wm_special_thread *item);
static int wm_currentthread_isspecial(void);
static unsigned int wm_head_isspecial(void);


// this variable hold the front of the queue
static struct wm_queue_item *queue_base;

// this variable hold the front of the list of special threads
static struct wm_special_thread *special_threads;

// a mutex to guard against hurting our queue of work items
static pthread_mutex_t queue_mutex;
static pthread_mutex_t special_mutex;
static pthread_cond_t special_item_cond;

int wm_init_manager(void) {
	queue_base = NULL;
	special_threads = NULL;
	
	pthread_cond_init(&wm_more_jobs, NULL);
	pthread_mutex_init(&wm_more_jobs_mutex, NULL);
	pthread_mutex_init(&queue_mutex, NULL);
	pthread_mutex_init(&special_mutex, NULL);
	pthread_cond_init(&special_item_cond, NULL);
	
	return 0;
}

void wm_destroy_manager(void) {
	wm_free_queueitem(queue_base);
	wm_free_specialitem(special_threads);
	
	queue_base = NULL;
	special_threads = NULL;
	
	pthread_cond_destroy(&wm_more_jobs);
	pthread_mutex_destroy(&wm_more_jobs_mutex);
	pthread_mutex_destroy(&queue_mutex);
	pthread_mutex_destroy(&special_mutex);
	pthread_cond_destroy(&special_item_cond);
}

int wm_register_special(pthread_t thread) {
	struct wm_special_thread *item;
	
	pthread_mutex_lock(&special_mutex);
	
	if (special_threads == NULL) {
		if ((special_threads = malloc(sizeof(struct wm_special_thread))) == NULL) {return -DFCLIENT_ENOMEM;}
		special_threads->thread = thread;
		special_threads->next = NULL;
	}
	else {
		item = special_threads;
		while (item->next != NULL) {item = item->next;}
		
		if ((item->next = malloc(sizeof(struct wm_special_thread))) == NULL) {return -DFCLIENT_ENOMEM;}
		item->next->thread = thread;
		item->next->next = NULL;
	}
	
	pthread_mutex_unlock(&special_mutex);
	
	return 0;
}



int wm_add_item(void *payload, unsigned int is_special) {
	struct wm_queue_item *item;
	
	// lock the queue
	pthread_mutex_lock(&queue_mutex);
	
	// find the first item with out a next
	// check if the base is null
	if (queue_base == NULL) {
		if ((queue_base = malloc(sizeof(struct wm_queue_item))) == NULL) {return -DFCLIENT_ENOMEM;}
		queue_base->payload = payload;
		queue_base->is_special = ((is_special > 0) ? 1 : 0);
		queue_base->next = NULL;
	}
	else {
		item = queue_base;
		while (item->next != NULL) {item = item->next;}
		
		if ((item->next = malloc(sizeof(struct wm_queue_item))) == NULL) {return -DFCLIENT_ENOMEM;}
		item->next->payload = payload;
		item->next->is_special = ((is_special > 0) ? 1 : 0);
		item->next->next = NULL;
	}
	
	// unlock the queue
	pthread_mutex_unlock(&queue_mutex);
	
	// alert anyone sitting on the condition var
	pthread_cond_broadcast(&wm_more_jobs);
	
	return 0;
}


void *wm_get_item(void){
	
	struct wm_queue_item *hold;
	char *ret;
	
	if (wm_head_isspecial() && !wm_currentthread_isspecial()) {
		// allow others to do some work
		pthread_mutex_unlock(&wm_more_jobs_mutex);
		
		// alert some other waiting thread
		pthread_cond_broadcast(&wm_more_jobs);
		
		// wait while the special thread does its thing
		pthread_cond_wait(&special_item_cond, &special_mutex);
		pthread_mutex_unlock(&special_mutex);

		return NULL;
	}
	
	// lock the queue
	pthread_mutex_lock(&queue_mutex);
	
	// check if the head is null
	if (queue_base== NULL) {
		pthread_mutex_unlock(&queue_mutex);
		return NULL;
	}
	
	// retrieve the current head and then set a new head
	hold = queue_base;
	ret = hold->payload;
	queue_base = hold->next;
	
	
	// unlock the queue
	pthread_mutex_unlock(&queue_mutex);
	
	if (hold->is_special) { pthread_cond_broadcast(&special_item_cond); }
	
	free(hold);
	
	return ret;
	
}

int wm_item_avaliable(void) { 
	int ret;
	
	// lock
	pthread_mutex_lock(&queue_mutex);
	
	ret = (queue_base!=NULL);
	
	// unlock
	pthread_mutex_unlock(&queue_mutex);
	
	return ret; 
}

void wm_dump(void) {
	int i = 0;
	struct wm_queue_item *current;
	
	current = queue_base;
	dferrors_printdebug("Job list\n-----------------\n");
	if (current==NULL) {dferrors_printdebug("(None)\n");}
	else {
		while (current!=NULL) {
			dferrors_printdebug("[%d] %p\n", i, current->payload);
		
			i++;
			current = current->next;
		}
	}
}



static void wm_free_queueitem(struct wm_queue_item *item) {
	if (item==NULL) return;
	if (item->next!=NULL) wm_free_queueitem(item->next);
	
	free(item->payload);
	free(item);
	
}

static void wm_free_specialitem(struct wm_special_thread *item) {
	if (item==NULL) return;
	if (item->next!=NULL) wm_free_specialitem(item->next);
	
	free(item);
	
}

static int wm_currentthread_isspecial() {
	struct wm_special_thread *current;
	pthread_t current_thread;
	
	// if there arent any special threads they are all
	// special to avoid a nasty deadlock
	if (special_threads == NULL) { return 1; }
	
	current_thread = pthread_self();
	current = special_threads;
	
	while (current!=NULL) {
		if (current->thread == current_thread) return 1;
		current = current->next;
	}
	
	return 0;
}

static unsigned int wm_head_isspecial() {
	return ((queue_base != NULL) && (queue_base->is_special == 1));
}





